package com.lintf.distributed.transaction.order.service.Impl;

import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.lintf.distributed.transaction.DeductInventoryEvent;
import com.lintf.distributed.transaction.order.entity.Message;
import com.lintf.distributed.transaction.order.entity.Order;
import com.lintf.distributed.transaction.order.mapper.OrderMapper;
import com.lintf.distributed.transaction.order.service.MessageService;
import com.lintf.distributed.transaction.order.service.OrderService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.math.BigDecimal;
import java.util.Date;
import java.util.UUID;

/**
 * @author lin tingfeng
 * @ system Mac
 * @ data 2025/6/12 17:33
 * @ package_name com.lintf.distributed.transaction.order.service
 * @ project_name distributed-transaction-demo
 * @ desc
 */
@Slf4j
@Service
public class OrderServiceImpl extends ServiceImpl<OrderMapper, Order> implements OrderService {

    private final MessageService messageService;
    private final KafkaTemplate<String, String> kafkaTemplate;

    @Autowired
    public OrderServiceImpl(MessageService messageService, KafkaTemplate<String, String> kafkaTemplate) {
        this.messageService = messageService;
        this.kafkaTemplate = kafkaTemplate;
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public String createOrder(Long userId, Long productId, Integer quantity, BigDecimal amount) {
        // 创建订单
        Order order = new Order();
        order.setUserId(userId);
        order.setProductId(productId);
        order.setQuantity(quantity);
        order.setAmount(amount);
        order.setStatus("PENDING");
        order.setCreatedAt(new Date());
        order.setUpdatedAt(new Date());
        save(order);

        log.info("订单创建成功，订单ID: {}", order.getId());

        // 准备扣减库存事件
        DeductInventoryEvent event = new DeductInventoryEvent();
        event.setOrderId(order.getId());
        event.setProductId(productId);
        event.setQuantity(quantity);

        // 发送事件到消息系统
        sendDeductInventoryEvent(event);

        return "订单创建成功，等待库存扣减";
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public void sendDeductInventoryEvent(DeductInventoryEvent event) {
        // 创建消息记录
        Message message = new Message();
        String messageId = UUID.randomUUID().toString();
        message.setMessageId(messageId);
        message.setContent(JSON.toJSONString(event));
        message.setQueueName("deduct-inventory");
        message.setStatus(0); // 未发送状态
        message.setRetryCount(0);

        messageService.save(message);
        log.info("库存扣减消息已存储到本地表，消息ID: {}", messageId);

        // 立即尝试发送消息
        trySendKafkaMessage(message);
    }

    private void trySendKafkaMessage(Message message) {
        try {
            // 发送消息到Kafka
            kafkaTemplate.send(message.getQueueName(), message.getMessageId(), message.getContent())
                    .thenAccept(result -> {
                        log.info("消息发送成功: topic={}, key={}",
                                result.getRecordMetadata().topic(),
                                message.getMessageId());
                        message.setStatus(1); // 标记为已发送
                        messageService.updateById(message);
                    })
                    .exceptionally(ex -> {
                        log.error("消息发送失败: {}", message.getMessageId(), ex);
                        messageService.incrementRetryCount(message);
                        return null;
                    });

        } catch (Exception e) {
            log.error("消息发送异常: {}", message.getMessageId(), e);
            messageService.incrementRetryCount(message);
        }
    }
}
